﻿using Confluent.Kafka;
using Elasticsearch.Net;
using Md.Demo.Entity.MongoModels;
using Microsoft.Extensions.Configuration;
using Nest;
using Newtonsoft.Json;
using Newtonsoft.Json.Converters;
using System;
using System.Collections.Generic;
using System.Linq;
using System.Threading.Tasks;

namespace Md.Demo.Tools
{
    public class ElasticSearchServer : IElasticSearchServer
    {
        public IElasticClient ElasticLinqClient { get; set; }

        public ElasticSearchServer()
        {
            //linq查询的初始化
            this.ElasticLinqClient = new ElasticClient(new Uri("http://47.96.31.8:9200/"));
        }

        #region 操作类
        public async Task<bool> IndexExistsAsync(string index)
        {
            return (await ElasticLinqClient.Indices.ExistsAsync(index)).Exists;
        }

        public async Task CreateAsync<T>(T entity, string index) where T : EsBase
        {
            //这里可判断是否存在
            var response = await ElasticLinqClient.IndexAsync(entity,
                s => s.Index(index));

            if (!response.IsValid)
                throw new Exception("新增数据失败:" + response.OriginalException.Message);
        }

        public async Task InsertRangeAsync<T>(IEnumerable<T> entity, string index) where T : EsBase
        {
            var bulkRequest = new BulkRequest(index)
            {
                Operations = new List<IBulkOperation>()
            };
            var operations = entity.Select(o => new BulkIndexOperation<T>(o)).Cast<IBulkOperation>().ToList();
            bulkRequest.Operations = operations;
            var response = await ElasticLinqClient.BulkAsync(bulkRequest);

            if (!response.IsValid)
                throw new Exception("批量新增数据失败:" + response.OriginalException.Message);
        }

        public async Task UpdateAsync<T>(T entity, string index) where T : EsBase
        {
            var response = await ElasticLinqClient.UpdateAsync<T>(entity.id, x => x.Index(index).Doc(entity));
            if (!response.IsValid)
                throw new Exception("更新失败:" + response.OriginalException.Message);
        }

        public async Task DeleteAsync<T>(string Id, string index) where T : EsBase
        {
            await ElasticLinqClient.DeleteAsync<T>(Id, x => x.Index(index));
        }

        public async Task RemoveIndex(string index)
        {
            var exists = await IndexExistsAsync(index);
            if (!exists) return;
            var response = await ElasticLinqClient.Indices.DeleteAsync(index);

            if (!response.IsValid)
                throw new Exception("删除index失败:" + response.OriginalException.Message);
        }

        public async Task<Tuple<int, IList<T>>> QueryAsync<T>(int page, int limit, string index) where T : EsBase
        {
            var query = await ElasticLinqClient.SearchAsync<T>(x => x.Index(index)
                                    .From((page - 1) * limit)
                                    .Size(limit)
                                    .Sort(x => x.Descending(v => v.createDate)));
            return new Tuple<int, IList<T>>(Convert.ToInt32(query.Total), query.Documents.ToList());
        }
        #endregion
    }
}